from datetime import datetime
from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
}

dag = DAG(
    'test_dag',
    default_args=default_args,
    description='Test DAG that echoes a message and fails',
    schedule=None,
    catchup=False,
    tags=['test'],
)

test_task = BashOperator(
    task_id='echo_and_fail',
    bash_command='echo "This is a test message" && exit 1',
    dag=dag,
)

os.environ['AIRFLOW__CORE__DEFAULT_TASK_RETRIES'] = '3'

def return_deserialized_task_after_serializng(retries):
    from datetime import datetime
    from airflow.sdk import DAG
    from airflow.providers.standard.operators.bash import BashOperator
    from airflow.serialization.serialized_objects import SerializedDAG
    from airflow.sdk.bases.operator import OPERATOR_DEFAULTS
    dag_0 = DAG(f'test_{retries}', default_args={'retries': retries}, start_date=datetime(2024, 1, 1))
    task_0 = BashOperator(task_id=f't{retries}', bash_command='echo', dag=dag_0)
    serialized_0 = SerializedDAG.to_dict(dag_0)
    task_data_0 = serialized_0["dag"]["tasks"][0]["__var"]
    client_defaults = serialized_0.get("client_defaults", {}).get("tasks", {})
    deserialized_dag_0 = SerializedDAG.from_dict(serialized_0)
    deserialized_task_0 = deserialized_dag_0.get_task(f't{retries}')
    
    return deserialized_task_0

ds_task_0 = return_deserialized_task_after_serializng(0)
ds_task_0.retries
Out[8]: 3

ds_task_1 = return_deserialized_task_after_serializng(1)
ds_task_1.retries
Out[16]: 1

ds_task_0 = return_deserialized_task_after_serializng(0)
ds_task_0.retries
Out[6]: 0
ds_task_1 = return_deserialized_task_after_serializng(1)
ds_task_1.retries
Out[7]: 1

from datetime import datetime
from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
}

dag = DAG(
    'test_dag',
    default_args=default_args,
    description='Test DAG that echoes a message and fails',
    schedule=None,
    catchup=False,
    tags=['test'],
)

test_task = BashOperator(
    task_id='echo_and_fail',
    bash_command='echo "This is a test message" && exit 1',
    dag=dag,
)